package com.beiding.jsonanalyzer.querylink;

import com.beiding.jsonanalyzer.analyze.Window;
import com.beiding.jsonanalyzer.analyze.WindowInstance;
import com.beiding.jsonanalyzer.flow.BranchNode;
import com.beiding.jsonanalyzer.flow.ValueNode;
import com.beiding.jsonanalyzer.flow.flowevents.BatchEndEvent;
import com.beiding.jsonanalyzer.flow.flowevents.BatchStartEvent;
import com.beiding.jsonanalyzer.flow.flowevents.DataNodeEvent;

public class WindowQueryLinkNode<W extends ValueNode> extends ComputeNode<W> {

    private Window window;

    public WindowQueryLinkNode(QueryLinkNode<W> pre, Window window) {
        super(pre);
        this.window = window;
    }

    @Override
    protected void bindByComputeNode(BranchNode<W> preResult, BranchNode<W> newBranchNode) {

        //绑定基础事件
        ComputeNodeUtils.bindBaseEvent(preResult, newBranchNode);

        //新建窗口

        WindowInstance instance = window.createInstance(o -> {
            newBranchNode.signDataEvent(new BatchStartEvent<>(o));
        }, o -> {
            newBranchNode.signDataEvent(new BatchEndEvent<>(o));
        });

        preResult.onDataEvent(BatchStartEvent.eventType(), event -> {
            instance.start();
        });

        preResult.onDataEvent(BatchEndEvent.eventType(), event -> {
            instance.end();
        });

        preResult.onDataEvent(DataNodeEvent.eventType(), event -> {
            instance.add(() -> {
                newBranchNode.signDataEvent(event);
            });
        });

    }
}
